package thread.disruptor.generate1;

import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.IgnoreExceptionHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.WorkHandler;
import com.lmax.disruptor.WorkerPool;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by Try on 2017/6/11.
 */
public class Main02 {
    public static void main(String[] args) throws InterruptedException {
        final int ringBufferSize = 1024;
        final int THREAD_NUMBERS = 4;
        final int loop = 8;

        CountDownLatch countDownLatch = new CountDownLatch(loop);

        EventFactory<Trade> eventFactory = new EventFactory<Trade>() {
            @Override
            public Trade newInstance() {
                return new Trade();
            }
        };

        RingBuffer<Trade> ringBuffer = RingBuffer.createSingleProducer(eventFactory, ringBufferSize);
        SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
        ExecutorService executorService = Executors.newFixedThreadPool(THREAD_NUMBERS);
        ExecutorService es = Executors.newFixedThreadPool(THREAD_NUMBERS);

        WorkHandler<Trade> handler = new TradeHandler(countDownLatch);
        WorkerPool<Trade> workerPool = new WorkerPool<Trade>(
                ringBuffer, sequenceBarrier, new IgnoreExceptionHandler(), handler);

        workerPool.start(executorService);
        es.execute(() -> {
            for (int i = 0; i < loop; i++) {
                long seq = ringBuffer.next();
                ringBuffer.get(seq).setPrice(Math.random() * 9999);
                ringBuffer.publish(seq);
            }
        });
        countDownLatch.await();
        workerPool.halt();
        executorService.shutdown();
        es.shutdown();
    }
}
